/*
* @Author: Rui XuLe
* @Date:   2019-07-12 22:40:03
* @Last Modified by:   Rui XuLe
* @Last Modified time: 2019-07-20 10:30:54
 */
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"math/rand"

	"time"

	"github.com/Shopify/sarama"
)

var Address = []string{"81.70.159.154:9092"}

const (
	TOPIC = "test_2021_11_27_rxl"
)

func main() {
	p := syncProducerInit(Address)
	sendMeg(p)
}

type MegInfo struct {
	Name    string    `json:"name"`
	Meg     string    `json:"meg"`
	TimeMeg time.Time `json:"timeTime"`
}

//同步消息模式
func syncProducerInit(address []string) (p sarama.SyncProducer) {
	//创建config 配置对象
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	//设置超时时间
	config.Producer.Timeout = 5 * time.Second
	//采用随机分区的分区策略
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	//采用随机分区
	sarama.NewRandomPartitioner(TOPIC)
	//同步生产者
	p, err := sarama.NewSyncProducer(address, config)
	if err != nil {
		log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
		return nil
	}
	return
}

//主要是 消息的发送
func sendMeg(p sarama.SyncProducer) {
	defer p.Close()
	//topic := "test20200618"
	m := MegInfo{
		Name:    "芮旭乐",
		Meg:     "今天是20200618，我在测试gokafka",
		TimeMeg: time.Now(),
	}
	data, _ := json.Marshal(m)
	//for i := 0; i < 100; i++ {
	for {
		value := string(data) // fmt.Sprintf(srcValue)
		msg := &sarama.ProducerMessage{
			Partition: int32(4),
			Key:       sarama.StringEncoder(fmt.Sprintf("%d", rand.Intn(10))),
			Timestamp: time.Now(),
			Topic:     TOPIC,
			//Value:     sarama.ByteEncoder(value),
			Value : sarama.StringEncoder(value),
		}
		part, offset, err := p.SendMessage(msg)
		if err != nil {
			log.Printf("send message(%s) err=%s \n", value, err)
		} else {
			log.Printf(" %v , 发送成功，partition=%d, offset=%d \n", value, part, offset)
		}
		time.Sleep(time.Second * 5)
	}
}
